package com.shanzhe.eshop.common.nio;

import org.apache.catalina.Server;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.*;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.nio.charset.CharsetEncoder;
import java.util.Iterator;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Author yinwei
 * @create 2021/12/25
 * @Version 1.0
 */
public class NIOServer {

    private static CharsetEncoder encoder =  Charset.forName("UTF-8").newEncoder();
    private static CharsetDecoder decoder =  Charset.forName("UTF-8").newDecoder();

    private static ByteBuffer readBuffer;
    private static Selector selector;
    private static LinkedBlockingQueue<SelectionKey> selectionKeysQueue;
    private static ThreadPoolExecutor threadPool;

    public static void main(String[] args) throws Exception{
        init();
        listen();
    }

    private static void init(){
        //读取请求数据的buffer缓冲
        readBuffer = ByteBuffer.allocateDirect(1024);
        //用于和客户端进行连接
        ServerSocketChannel serverSocketChannel = null;

        try {
            //NIO多路复用的组件
            selector = Selector.open();

            serverSocketChannel = ServerSocketChannel.open();
            //NIO支持非阻塞
            serverSocketChannel.configureBlocking(false);
            //负责和各个客户端建立连接请求
            serverSocketChannel.socket().bind(new InetSocketAddress(9000),100);

            //将serverSocketChannel注册到selector上 仅仅关注这个serverSocketChannel接收到的tcp连接请求
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

            selectionKeysQueue = new LinkedBlockingQueue<SelectionKey>(500);
            threadPool = new ThreadPoolExecutor(10, 10, 10, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<Runnable>(), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());

            for(int i = 0;i<10;i++){
                threadPool.submit(new Worker());
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    private static void listen(){
        while (true){
            try {
                //进行阻塞 如果有客户端发送tcp连接请求到serverSocketChannel中就会释放
                selector.select();
                //从selector中拿到selectedKeys 一个SelectionKey简单理解为代表一个连接请求
                Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();

                while (keyIterator.hasNext()){
                    SelectionKey key = keyIterator.next();
                    keyIterator.remove();
                    selectionKeysQueue.offer(key);
                }
            }catch (Exception e){

            }
        }
    }


    static class Worker implements Runnable{
        @Override
        public void run() {
            while (true){
                try {
                    SelectionKey key = selectionKeysQueue.take();
                    handleRequest(key);
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }

        private static void handleRequest(SelectionKey key) throws IOException {
            SocketChannel socketChannel = null;

            try {
                if(key.isAcceptable()){
                    //如果key是连接请求
                    ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
                    //accept就可以和客户端进行tcp的三次握手
                    socketChannel = serverSocketChannel.accept();
                    if(socketChannel!=null){
                        socketChannel.configureBlocking(false);
                        socketChannel.register(selector,SelectionKey.OP_READ);
                    }
                }else if(key.isReadable()){
                    //如果key是发送数据请求
                    socketChannel = (SocketChannel) key.channel();
                    readBuffer.clear();
                    int count = socketChannel.read(readBuffer);
                    if(count>0){
                        //将buffer的position进行复位 position = 0 limit = 字节为
                        readBuffer.flip();
                        CharBuffer charBuffer = decoder.decode(readBuffer);
                        String request = charBuffer.toString();
                        System.out.println(request);

                        socketChannel.register(selector,SelectionKey.OP_WRITE);
                    }else{
                        socketChannel.close();
                    }
                }else if(key.isWritable()){
                    socketChannel = (SocketChannel) key.channel();
                    ByteBuffer writeBuffer = ByteBuffer.allocateDirect(1024);
                    String response = "响应数据";
                    writeBuffer.put(response.getBytes());

                    socketChannel.write(writeBuffer);
                    socketChannel.register(selector,SelectionKey.OP_READ);
                }
            }catch (Exception e){
                e.printStackTrace();
                if(socketChannel!=null){
                    socketChannel.close();
                }
            }
        }
    }


}
